/**
 * Copyright (C) 2017-2018 pinganfang, Inc. All Rights Reserved.
 */
package com.pingan.haofang.searchcloud.solr.collection.impl;

import com.alibaba.fastjson.JSON;
import com.pingan.haofang.searchcloud.api.CollectionManagerService;
import com.pingan.haofang.searchcloud.api.CollectionSummaryDto;
import com.pingan.haofang.searchcloud.api.DiskConditonDTO;
import com.pingan.haofang.searchcloud.api.constants.EngineType;
import com.pingan.haofang.searchcloud.api.matedata.CollectionMetaData;
import com.pingan.haofang.searchcloud.solr.api.client.SolrClientFactory;
import com.pingan.haofang.searchcloud.solr.constants.SolrConstant;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.impl.ZkClientClusterStateProvider;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.CoreAdminRequest;
import org.apache.solr.client.solrj.response.CollectionAdminResponse;
import org.apache.solr.client.solrj.response.CoreAdminResponse;
import org.apache.solr.common.params.CoreAdminParams;
import org.apache.solr.common.util.SimpleOrderedMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @author huhailiang937
 * @date 2017年4月6日下午9:16:49
 * @since 1.0.0
 */
@Service
@ManagedResource
public class SolrCollectionManagerService implements CollectionManagerService {

    private static final Pattern MAX_NODE_SET_PATTERN = Pattern.compile("the number of nodes currently live or live and part of your createNodeSet is (\\d)\\.");
    protected static Logger LOG = LoggerFactory.getLogger(SolrCollectionManagerService.class);

    private static final String CORE_TEMPLATE = "%s_shard%d_replica%d";

    @Autowired
    protected SolrClientFactory clientFactory;

    @SuppressWarnings({"rawtypes"})
    @Override
    public List<CollectionMetaData> list(String zkHosts) throws Exception {

        CollectionAdminRequest.ClusterStatus clusterStatus = new CollectionAdminRequest.ClusterStatus();

        CloudSolrClient solrClient = clientFactory.getByHost(zkHosts);

        CollectionAdminResponse statusResponse = clusterStatus.process(solrClient);
        SimpleOrderedMap map = (SimpleOrderedMap) statusResponse.getResponse().get("cluster");
        SimpleOrderedMap collectionsMap = (SimpleOrderedMap) map.get("collections");

        List<CollectionMetaData> collections = new ArrayList<CollectionMetaData>();
        for (Object entry : collectionsMap.asMap(0).keySet()) {
            String indexName = entry.toString();
            LinkedHashMap indexMap = (LinkedHashMap) collectionsMap.get(indexName);
            CollectionMetaData collection = JSON.parseObject(JSON.toJSONString(indexMap), CollectionMetaData.class);
            collection.setName(indexName);
            LinkedHashMap shardsMap = (LinkedHashMap) indexMap.get("shards");
            collection.setShardCount(shardsMap.size());
            collections.add(collection);
        }
        return collections;
    }

    @SuppressWarnings({"rawtypes"})
    @Override
    public CollectionMetaData get(String zkHosts, String collectionName) throws Exception {
        Assert.notNull(collectionName, "collectionName can not null");
        CollectionAdminRequest.ClusterStatus clusterStatus = new CollectionAdminRequest.ClusterStatus();
        clusterStatus.setCollectionName(collectionName);

        CloudSolrClient solrClient = clientFactory.getByHost(zkHosts);

        CollectionAdminResponse statusResponse = clusterStatus.process(solrClient);
        SimpleOrderedMap map = (SimpleOrderedMap) statusResponse.getResponse().get("cluster");
        SimpleOrderedMap collectionsMap = (SimpleOrderedMap) map.get("collections");

        LinkedHashMap indexMap = (LinkedHashMap) collectionsMap.get(collectionName);
        CollectionMetaData collection = JSON.parseObject(JSON.toJSONString(indexMap), CollectionMetaData.class);
        collection.setName(collectionName);
        LinkedHashMap shardsMap = (LinkedHashMap) indexMap.get("shards");
        collection.setShardCount(shardsMap.size());

        return collection;
    }

    @SuppressWarnings("deprecation")
    @Override
    public boolean create(String zkHost, CollectionMetaData solrCollection) throws Exception {
        Assert.notNull(solrCollection, "solrCollection can not null");
        Assert.notNull(solrCollection.getName(), "solrCollection.getName can not null");
        Assert.notNull(solrCollection.getConfigName(), "solrCollection.getConfigName can not null");
        Assert.notNull(solrCollection.getShardCount(), "solrCollection.getShardCount can not null");
        Assert.notNull(solrCollection.getReplicationFactor(), "solrCollection.getReplicationFactor can not null");

        CollectionAdminRequest.Create create = new CollectionAdminRequest.Create();
        create.setCollectionName(solrCollection.getName());
        create.setConfigName(solrCollection.getConfigName());
        create.setNumShards(solrCollection.getShardCount());
        create.setReplicationFactor(solrCollection.getReplicationFactor());


        if (null == solrCollection.getRouter()) {
            create.setRouterName("compositeId");
        } else {
            create.setRouterName(solrCollection.getRouter().getName());
        }

        if (null != solrCollection.getMaxShardsPerNode()) {
            create.setMaxShardsPerNode(solrCollection.getMaxShardsPerNode());
        }
        try {
            CloudSolrClient solrClient = clientFactory.getByHost(zkHost);

            CollectionAdminResponse response = create.process(solrClient);
            return (null != response) && response.isSuccess();
        } catch (Exception e) {
            handleCreateException(e);
            return false;
        }
    }

    /**
     * 处理异常
     *
     * @param e
     * @throws Exception
     */
    private void handleCreateException(Exception e) throws Exception {
        Matcher matcher = MAX_NODE_SET_PATTERN.matcher(e.getMessage());
        if (matcher.find()) {
            String maxRelicaFactor = matcher.group(1);
            throw new CollectionCreateException(Integer.parseInt(maxRelicaFactor));
        } else {
            throw e;
        }
    }


    @SuppressWarnings("deprecation")
    @Override
    public boolean delete(String zkHost, String collectionName) throws Exception {
        CollectionAdminRequest.Delete delete = new CollectionAdminRequest.Delete();
        delete.setCollectionName(collectionName);

        CloudSolrClient solrClient = clientFactory.getByHost(zkHost);

        CollectionAdminResponse response = delete.process(solrClient);
        return (null == response) ? false : response.isSuccess();
    }

    @Override
    @ManagedOperation
    public List<String> getAllCollectionNames(String clusterHost) throws Exception {
        CloudSolrClient solrClient = clientFactory.getByHost(clusterHost);
        return CollectionAdminRequest.listCollections(solrClient);
    }

    @Override
    public Map<String, CollectionSummaryDto> getAllCollectionSummary(String clusterHost, List<CollectionMetaData> list) throws Exception {
        List<String> liveHostUrl = this.getLiveHostUrl(clusterHost);
        //遍历host拿到所有副本
        Map<String, CollectionSummaryDto> replicationDtoMap = new HashMap<>();
        for (String host : liveHostUrl) {
            LOG.info("start to get solr core summary from host:{}", host);
            replicationDtoMap.putAll(getCollectionReplicationByHost(host));
        }
        return buildIndexSummary(replicationDtoMap, list);
    }

    /**
     * 将collection和core信息进行分组计算
     *
     * @param replicationDtoMap      solr core map
     * @param collectionMetaDataList solrCollectionMetadataList
     * @return
     */
    private Map<String, CollectionSummaryDto> buildIndexSummary(Map<String, CollectionSummaryDto> replicationDtoMap,
                                                                List<CollectionMetaData> collectionMetaDataList) {

        if (CollectionUtils.isEmpty(collectionMetaDataList)) {
            return Collections.emptyMap();
        }

        Map<String, CollectionSummaryDto> collectionSummaryMap = new HashMap<>(collectionMetaDataList.size());
        for (CollectionMetaData data : collectionMetaDataList) {
            Integer shardCount = data.getShardCount();
            Integer replicationCount = data.getReplicationFactor();
            String collectionName = data.getName();
            CollectionSummaryDto summaryDto = new CollectionSummaryDto(collectionName);
            long totalNum = 0;
            for (int i = 1; i <= shardCount; i++) {
                long numDocs = 0;
                for (int j = 1; j <= replicationCount; j++) {
                    String replicationName = String.format(CORE_TEMPLATE, data.getName(), i, j);
                    CollectionSummaryDto replicaSummaryDto = replicationDtoMap.get(replicationName);
                    if (Objects.isNull(replicaSummaryDto)) {
                        LOG.warn("solr collection :[{}] has miss a replication:[{}] ", collectionName, replicationName);
                        continue;
                    }
                    summaryDto.setSizeInKBytes(summaryDto.getSizeInKBytes() + replicaSummaryDto.getSizeInKBytes());
                    numDocs = replicaSummaryDto.getNumDocs();
                }
                totalNum += numDocs;
            }
            summaryDto.setNumDocs(totalNum);
            collectionSummaryMap.put(collectionName, summaryDto);
        }
        return collectionSummaryMap;
    }

    /**
     * 获取集群中的host url
     *
     * @param clusterHost
     * @return
     */
    @Override
    public List<String> getLiveHostUrl(String clusterHost) {
        ZkClientClusterStateProvider clusterStateProvider = (ZkClientClusterStateProvider) clientFactory.getByHost(clusterHost).getClusterStateProvider();
        Set<String> hostNodes = clusterStateProvider.liveNodes();
        return hostNodes.stream().map(host -> String.join(StringUtils.EMPTY, SolrConstant.HTTP_URL_PREFIX, host.replace("_", "/")))
                .collect(Collectors.toList());
    }

    /**
     * 获取一个集群下 存活节点的磁盘占用量
     *
     * @param clusterHost 集群host
     * @return
     */
    @Override
    public Map<String, DiskConditonDTO> getNodesSummary(String clusterHost) throws Exception {

        Map<String, DiskConditonDTO> map = new HashMap<>(16);

        List<String> liveHostUrls = this.getLiveHostUrl(clusterHost);
        LOG.info("start to get solr core summary from host:{}", clusterHost);
        for (String host : liveHostUrls) {
            // 获得每个host的磁盘存储信息
            Map<String, CollectionSummaryDto> summaryDtoMap = getCollectionReplicationByHost(host);

            DiskConditonDTO diskConditonDTO = new DiskConditonDTO();
            long disk_occupancy = 0;
            for (CollectionSummaryDto value : summaryDtoMap.values()) {
                disk_occupancy += value.getSizeInKBytes();
            }
            // 转换成MB
            diskConditonDTO.setDiskUsed(disk_occupancy / 1024);
            map.put(getIP(host), diskConditonDTO);
        }
        return map;

    }

    /**
     * 获得ip+端口号10.59.78.22:8983
     */
    public static String getIP(String url) {
        // 使用正则表达式过滤，
        String re =
                "((http|ftp|https)://)(([a-zA-Z0-9._-]+)|([0-9]{1,3}.[0-9]{1,3}.[0-9]{1,3}.[0-9]{1,3}))(([a-zA-Z]{2,6})|(:[0-9]{1,4})?)";
        String str = "";
        // 编译正则表达式
        Pattern pattern = Pattern.compile(re);
        // 忽略大小写的写法
        // Pattern pat = Pattern.compile(regEx, Pattern.CASE_INSENSITIVE);
        Matcher matcher = pattern.matcher(url);
        // 若url==http://127.0.0.1:9040或www.baidu.com的，正则表达式表示匹配
        if (matcher.matches()) {
            str = url;
        } else {
            String[] split2 = url.split(re);
            if (split2.length > 1) {
                String substring = url.substring(0, url.length() - split2[1].length());
                str = substring;
            } else {
                str = split2[0];
            }
        }
        return str.split("//")[1];
    }

    /**
     * 查询单个host的collection 信息
     *
     * @param hostUrl
     * @return map.key：solr的replication名字，map.value: {@link CollectionSummaryDto}
     * @throws IOException
     * @throws SolrServerException
     */
    @SuppressWarnings("deprecation")
    private Map<String, CollectionSummaryDto> getCollectionReplicationByHost(String hostUrl) throws IOException, SolrServerException {
        HttpSolrClient solrClient = new HttpSolrClient(hostUrl);
        solrClient.setConnectionTimeout(SolrConstant.TIME_OUT_TEN_SECOND);
        CoreAdminRequest coreAdminRequest = new CoreAdminRequest();
        coreAdminRequest.setAction(CoreAdminParams.CoreAdminAction.STATUS);
        CoreAdminResponse process = coreAdminRequest.process(solrClient);
        Map<String, LinkedHashMap<String, Object>> map = process.getCoreStatus().asMap(3);
        LOG.info(" request solr core result :{}", map);
        Map<String, CollectionSummaryDto> collect = map.entrySet().stream().map(Map.Entry::getValue).map(this::conv2SummaryDto)
                .collect(Collectors.toMap(CollectionSummaryDto::getName, Function.identity(), (o, n) -> (n)));
        LOG.info("request solr core result size :{}", collect.size());
        return collect;
    }

    private CollectionSummaryDto conv2SummaryDto(Map<String, Object> map) {
        CollectionSummaryDto dto = new CollectionSummaryDto();
        dto.setName(map.get("name").toString());
        Map index = (Map) map.get("index");
        dto.setNumDocs(((Integer) index.get("numDocs")).longValue());
        dto.setSizeInKBytes(conv2KBytes((long) index.get("sizeInBytes")));
        dto.setSize(index.get("size").toString());
        return dto;
    }

    private long conv2KBytes(Long bytes) {
        return BigDecimal.valueOf(bytes).divide(BigDecimal.valueOf(SolrConstant.BYTE_UNIT),
                0, BigDecimal.ROUND_HALF_UP).longValue();
    }

    @Override
    public EngineType getEngineType() {
        return EngineType.SOLR;
    }


}
